package org.apache.flink.runtime.messages.webmonitor;

import javax.annotation.Nullable;

import org.apache.flink.configuration.MemorySize;
import org.apache.flink.runtime.resourcemanager.ResourceOverview;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonCreator;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty;

/**
 * Response to the {@link RequestStatusOverview} message, carrying a description of the Flink
 * cluster status.
 */
public class ClusterOverview implements InfoMessage {

    private static final long serialVersionUID = -729861859715105265L;

    public static final String FIELD_NAME_RPS = "resource-providers";
    public static final String FIELD_NAME_SLOTS_TOTAL = "slots-total";
    public static final String FIELD_NAME_SLOTS_AVAILABLE = "slots-available";
    public static final String FIELD_NAME_SLOTS_ALLOCATED = "slots-allocated";
    public static final String FIELD_NAME_CPUCORE_TOTAL = "cpucore-total";
    public static final String FIELD_NAME_MEMORY_SIZE = "memory-size";

    @JsonProperty(FIELD_NAME_RPS)
    private final int numResourcePoviders;

    @JsonProperty(FIELD_NAME_SLOTS_TOTAL)
    private final int numSlotsTotal;

    /** 可用资源slot*/
    @JsonProperty(FIELD_NAME_SLOTS_AVAILABLE)
    private final int numSlotsAvailable;
    
    @JsonProperty(FIELD_NAME_SLOTS_ALLOCATED)
    private final int numSlotAllocated;
    
	/** cpu core数量*/
    @JsonProperty(FIELD_NAME_CPUCORE_TOTAL)
	private final int numCpuCoreTotal;
	
	/** 堆内存总数量*/
    @JsonProperty(FIELD_NAME_MEMORY_SIZE)
	private final MemorySize numHeapMemoryTotal;

    @JsonCreator
    // numTaskManagersBlocked and numSlotsFreeAndBlocked is Nullable since Jackson will assign null
    // if the field is absent while parsing
    public ClusterOverview(
            @JsonProperty(FIELD_NAME_RPS) int numResourcePoviders,
            @JsonProperty(FIELD_NAME_SLOTS_TOTAL) int numSlotsTotal,
            @JsonProperty(FIELD_NAME_SLOTS_AVAILABLE) int numSlotsAvailable,
            @JsonProperty(FIELD_NAME_CPUCORE_TOTAL) int numCpuCoreTotal,
            @JsonProperty(FIELD_NAME_CPUCORE_TOTAL) @Nullable MemorySize numHeapMemoryTotal
            ) {
        this.numResourcePoviders = numResourcePoviders;
        this.numSlotsTotal = numSlotsTotal;
        this.numSlotsAvailable = numSlotsAvailable;
        this.numCpuCoreTotal = numCpuCoreTotal;
        this.numHeapMemoryTotal = numHeapMemoryTotal;
        this.numSlotAllocated = numSlotsTotal - numSlotsAvailable;

    }

    public int getNumSlotsTotal() {
        return numSlotsTotal;
    }

    public int getNumSlotsAvailable() {
        return numSlotsAvailable;
    }

    public int getNumResourcePoviders() {
		return numResourcePoviders;
	}

	public int getNumSlotAllocated() {
		return numSlotAllocated;
	}

	public int getNumCpuCoreTotal() {
		return numCpuCoreTotal;
	}

	public MemorySize getNumHeapMemoryTotal() {
		return numHeapMemoryTotal;
	}

	@Override
    public boolean equals(Object obj) {
        if (this == obj) {
            return true;
        } else if (obj instanceof ClusterOverview) {
            ClusterOverview that = (ClusterOverview) obj;
            return this.numResourcePoviders == that.numResourcePoviders
                    && this.numSlotsTotal == that.numSlotsTotal
                    && this.numSlotsAvailable == that.numSlotsAvailable
                    && this.numCpuCoreTotal == that.numCpuCoreTotal
                    && this.numHeapMemoryTotal.equals(that.numHeapMemoryTotal);
        } else {
            return false;
        }
    }

    @Override
    public int hashCode() {
        int result = super.hashCode();
        result = 31 * result + numResourcePoviders;
        result = 31 * result + numSlotsTotal;
        result = 31 * result + numSlotsAvailable;
        result = 31 * result + numCpuCoreTotal;
        result = 31 * result + numHeapMemoryTotal.hashCode();
        return result;
    }
    
	//
	public static ClusterOverview from(ResourceOverview resourceOverview) {
		
	    int numResourcePoviders = resourceOverview.getNumberWorkers();
		//
		int numSlotTotal = resourceOverview.getNumberRegisteredSlots();
		// 
		int numSlotAllocated = resourceOverview.getNumberFreeSlots();
		//
		int numSlotsAvailable = numSlotTotal - numSlotAllocated;
				
		int numCpuCoreTotal = resourceOverview.getTotalResource().getCpuCores().getValue().intValue();
		/** 堆内存总数量*/
		MemorySize numHeapMemoryTotal = resourceOverview.getTotalResource().getTaskHeapMemory();
	    /** 资源提供者数量*/

		return new ClusterOverview(numResourcePoviders, numSlotTotal,  numSlotsAvailable, 
				numCpuCoreTotal, numHeapMemoryTotal);
		
	}
}
